package com.ylkj.bus.dispatcher.handler;

import com.hwl.datahub.DataHubWriteConfig;
import com.hwl.datahub.out.ProxyDatahubSinkFunctionStream;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.device.report.DeviceOnlineDataHandler;
import com.ylkj.bus.device.report.UpdateDevicReportAndUnpack;
import com.ylkj.bus.dispatcher.IStreamHandler;

import com.ylkj.stream.datahub.SinkResolver;
import com.ylkj.stream.rds.RdsConfig;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * @Author: wengzx
 * @Date: 2025/4/18
 * @Desc:
 */
@RequiredArgsConstructor
@Data
public class DeviceOnlineDataStreamHandler implements IStreamHandler {
    private final RdsConfig rdsConfig;
    private final String rdsTableName;
    private final DataHubWriteConfig dataHubSinkConfig;

    @Override
    public void handle(DataStream<MessageContent> stream) {
        SingleOutputStreamOperator<String> deviceOnlineProcessData = stream.process(new DeviceOnlineDataHandler()).name("collect_process");
        //deviceOnlineProcessData.print();
        deviceOnlineProcessData.addSink(new ProxyDatahubSinkFunctionStream<>(
                dataHubSinkConfig,
                new SinkResolver()
        )).name(dataHubSinkConfig.getTopicName() + "_datahub");
        //deviceOnlineProcessData.process(new UpdateDevicReportAndUnpack(rdsTableName, rdsConfig)).name("rds_update_insert_process");
    }
}
